package com.bfxy.kafka.api.producer;

import com.alibaba.fastjson.JSON;
import com.bfxy.kafka.api.constant.Const;
import com.bfxy.kafka.api.model.User;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.errors.LeaderNotAvailableException;
import org.apache.kafka.common.errors.NetworkException;
import org.apache.kafka.common.errors.RecordTooLargeException;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.Future;

public class NormalProducer {

    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.218.21:9092");
        properties.put(ProducerConfig.CLIENT_ID_CONFIG, "normal-producer");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // kafka消息的重试机制：RETRIES_CONFIG参数默认是0
        // 作用是如果发送失败，判断是否属于可重试异常，如果可以重试，则放入队列等待再次轮询，如果不可以重试则抛出异常，由业务方catch处理
        properties.put(ProducerConfig.RETRIES_CONFIG, 3);

        // 可重试异常
        // org.apache.kafka.common.errors.NetworkException：比如网络产生一些异常抖动没有收到请求或者响应，或者消息没发出去
        // org.apache.kafka.common.errors.LeaderNotAvailableException：比如当前一刹那kafka集群没有leader，可能正在选举的过程中

        // 不可重试异常
        // org.apache.kafka.common.errors.RecordTooLargeException：消息体过大

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(properties)) {
            User user = new User("001", "xiao xiao");
            ProducerRecord<String, String> record = new ProducerRecord<>(Const.TOPIC_NORMAL, JSON.toJSONString(user));
            /**
             * 一条消息必须通过key去计算出来实际的partition，按照partition去存储的
             * 新创建消息：ProducerRecord(
             * topic=topic_normal,
             * partition=null,
             * headers=RecordHeaders(headers = [], isReadOnly = false),
             * key=null,
             * value={"id":"001","name":"xiao xiao"},
             * timestamp=null)
             * 可以看到新创建出来的消息的key和partition是空的，所以必定是在下面发送消息的send()方法中计算得到的partition，可以通过查看源码发现它是通过一些算法计算出来的
             */
            System.out.println("新创建消息：" + record);
            // 一个参数的send方法本质上也是异步的，因为这个方法返回的是一个java.util.concurrent.Future对象
            // 下面这种调用一个参数send()方法后直接调用get()方法，可以认为这是同步阻塞的
//            Future<RecordMetadata> future = producer.send(record);
//            RecordMetadata recordMetadata = future.get();
//            System.out.println(String.format("分区：%s，偏移量：%s，时间戳：%s", recordMetadata.partition(), recordMetadata.offset(), recordMetadata.timestamp()));

            // 带有两个参数send()方法是完全异步的，在回调Callback()方法中得到发送消息的结果
            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if (null != e) {
                        e.printStackTrace();
                    }
                    System.out.println(String.format("分区：%s，偏移量：%s，时间戳：%s", recordMetadata.partition(), recordMetadata.offset(), recordMetadata.timestamp()));
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
